package c.com.simple.learning.redis.delay;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import org.springframework.util.CollectionUtils;
import redis.clients.jedis.Jedis;

import java.lang.reflect.Type;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

public class DelayQueue<T> {

    private final Type taskType = new TypeReference<TaskItem<T>>() {
    }.getType();

    private final Jedis jedis;
    private final String queueKey;

    public DelayQueue(Jedis jedis, String queueKey) {
        this.jedis = jedis;
        this.queueKey = queueKey;
    }


    public static void main(String[] args) throws InterruptedException {
        Jedis jedis1 = new Jedis("127.0.0.1", 6379);
        Jedis jedis2 = new Jedis("127.0.0.1", 6379);
        Jedis jedis3 = new Jedis("127.0.0.1", 6379);
        String queueKey1 = "queue1";

        DelayQueue<String> queue1 = new DelayQueue<>(jedis1, queueKey1);
        DelayQueue<String> queue2 = new DelayQueue<>(jedis2, queueKey1);
        DelayQueue<String> queue3 = new DelayQueue<>(jedis3, queueKey1);

        Thread producer = new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                queue1.delay("这是一条测试数据~~" + i);
                try {
                    TimeUnit.MILLISECONDS.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"producer");

        Thread consumer1 = new Thread(queue2::loop,"consumer1");
        Thread consumer2 = new Thread(queue3::loop,"consumer2");

        producer.start();

        consumer1.start();
        consumer2.start();


        producer.join();
        TimeUnit.MILLISECONDS.sleep(10);
        consumer1.interrupt();
        consumer1.join();
        consumer2.interrupt();
        consumer2.join();

    }

    public void delay(T msg) {
        String id = UUID.randomUUID().toString();
        TaskItem<T> taskItem = TaskItem.from(id, msg);
        String s = JSON.toJSONString(taskItem);
        long interval = System.currentTimeMillis() + 5000;
        jedis.zadd(queueKey, interval, s);
    }

    public void loop() {
        while (!Thread.interrupted()) {
            List<String> messages = jedis.zrangeByScore(queueKey, 0, System.currentTimeMillis(), 0, 1);
            if (CollectionUtils.isEmpty(messages)) {
                try {
                    TimeUnit.MILLISECONDS.sleep(500);
                } catch (InterruptedException e) {
                    break;
                }
                continue;
            }

            String next = messages.iterator().next();
            if (jedis.zrem(queueKey, next) > 0) {
                TaskItem<T> taskItem = JSON.parseObject(next, taskType);
                handleMsg(taskItem.getMsg());
            }
        }
        System.out.println("-- end --");
    }

    private void handleMsg(T m) {
        System.out.println(Thread.currentThread().getName() + " " +m);
    }

    static class TaskItem<T> {

        public String id;
        public T msg;

        public static <T> TaskItem<T> from(String id, T msg) {
            TaskItem<T> taskItem = new TaskItem<>();
            taskItem.id = id;
            taskItem.msg = msg;
            return taskItem;
        }

        public String getId() {
            return id;
        }

        public void setId(String id) {
            this.id = id;
        }

        public T getMsg() {
            return msg;
        }

        public void setMsg(T msg) {
            this.msg = msg;
        }
    }
}